package org.apache.ocean;

import java.io.InputStreamReader;
import java.net.URL;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.FieldSelector;
import org.apache.lucene.document.SetBasedFieldSelector;
import org.apache.lucene.index.CorruptIndexException;
import org.apache.lucene.search.Query;
import org.apache.ocean.name.Node;
import org.apache.ocean.name.NodeNameClient;
import org.apache.ocean.server.Highlightings;

/**
 * 
 * 
 * @author jason rutherglen
 */
// TODO: test if text is being analyzed in the document
// TODO: make reloads delay period configurable
// TODO: do highlighting in mergeservice
// TODO: for shard moving add transactionrecord status
// TODO: compress document in transactionrecord
// TODO: implement UDP notification
// TODO: test node master slave replication
// TODO: get tomcat logs working, log master slave switches in separate log
public class SearchService implements Service {
	public static Logger log = Logger.getLogger(SearchService.class.getName());
	public static Logger seriousLog = Logger.getLogger("org.apache.solr.cluster.serious");
	public static long ADD_RAM_INDEX_DELAY = 1000 * 30;
	public static long LEASE_DURATION = 30 * 1000;
	public static int REPLICATION_HTTP_TIMEOUT = 4 * 1000;
	public static int NAME_HTTP_TIMEOUT = 5 * 1000;
	public SearchConfig searchConfig;
	public SearchIndexShardManager searchIndexShardManager;
	public ThreadPoolExecutor searchThreadPool;
	public ThreadPoolExecutor updateThreadPool;
	public ThreadPoolExecutor queryRewriteThreadPool;
	public ThreadPoolExecutor termFrequencyThreadPool;
	public ScheduledExecutorService timer;
	public NodeNameClient nodeNameClient;
	public ReplicationClient replicationClient;
	private String thisNodeID;
	private String reportableStatus;
	public Date thisNodeTimestamp;
	private String serviceStatus;
	public static FieldSelector idFieldSelector;
	public static FieldSelector allFieldSelector;
	public static FieldSelector headerFieldSelector;
	public static Set<String> headerSet;
	public static Set<String> defaultDocumentFieldCaches = new HashSet<String>(3);
	private NodeTransactionPushManager nodeTransactionPushManager = new NodeTransactionPushManager();
	//private RestartableTimer getShardInfosTimer;
	private ReentrantLock checkMasterSlaveStatusLock = new ReentrantLock();
	private CSequence cellTransactionIDSequence;

	static {
		defaultDocumentFieldCaches.add(ClusterUtil.ID_FIELD);
		defaultDocumentFieldCaches.add(ClusterUtil.VERSION_FIELD);
		defaultDocumentFieldCaches.add(ClusterUtil.SHARD_ID_FIELD);
	}

	public void resetSequences() throws Exception {
		TransactionID cellTransactionID = getHighestCellTransactionIDDatabase();
		if (cellTransactionID != null) {
			TransactionID clone = (TransactionID)cellTransactionID.clone();
			clone.increment();
			cellTransactionIDSequence.set(clone.value);
		}
	}

	public TransactionID getHighestCellTransactionIDDatabase() throws Exception {
		List<SearchIndexShard> searchIndexShards = searchIndexShardManager.getSearchIndexShards();
		List<TransactionID> list = new ArrayList<TransactionID>();
		for (SearchIndexShard searchIndexShard : searchIndexShards) {
			TransactionID transactionID = searchIndexShard.getHighestCellTransactionIDDatabase();
			if (transactionID != null) list.add(transactionID);
		}
		return ClusterUtil.max(list);
	}

	public void removeNodeFromBadTransactionPushList(String nodeid) {
		nodeTransactionPushManager.notBad(nodeid);
	}

	// TODO: have slave turn back on if it wants to
	public static class NodeTransactionPushManager {
		private Map<String, NodeStats> badNodeMap = new HashMap<String, NodeStats>();
		private ReentrantLock lock = new ReentrantLock();

		public void notBad(String nodeid) {
			lock.lock();
			try {
				badNodeMap.remove(nodeid);
			} finally {
				lock.unlock();
			}
		}

		public static class NodeStats {
			public Node node;
			public CLong failures;

			public NodeStats(Node node) {
				this.node = node;
				this.failures = new CLong(1);
			}

			public boolean isBad() {
				if (failures.value >= 3) {
					return true;
				} else
					return false;
			}
		}

		public void addBadNode(Node badNode) {
			lock.lock();
			try {
				NodeStats nodeStats = badNodeMap.get(badNode.id);
				if (nodeStats == null) {
					nodeStats = new NodeStats(badNode);
					badNodeMap.put(badNode.id, nodeStats);
				} else {
					nodeStats.failures.value++;
				}
			} finally {
				lock.unlock();
			}
		}

		public List<Node> getNodes(List<Node> incomingNodes) {
			if (incomingNodes == null)
				return null;
			lock.lock();
			try {
				List<Node> goodNodes = new ArrayList<Node>();
				for (Node incomingNode : incomingNodes) {
					NodeStats badNodeStats = badNodeMap.get(incomingNode.id);
					if (!badNodeStats.isBad()) {
						goodNodes.add(incomingNode);
					}
				}
				return goodNodes;
			} finally {
				lock.unlock();
			}
		}
	}

	// TODO: figure out way to turn pushNewTransactions back on in the event of a
	// timeout
	public void newTransactionRecord(TransactionRecord transactionRecord) throws Exception {
		List<Node> nodes = nodeTransactionPushManager.getNodes(nodeNameClient.nodesToCommitTransactionTo.get());
		if (nodes != null) {
			for (Node node : nodes) {
				try {
					HttpTimeout httpTimeout = new HttpTimeout(1000 * 2, 1000 * 5, new TimeoutState(6 * 1000));
					NodeClient.addTransactionRecord(node.url, transactionRecord, httpTimeout);
				} catch (GeneralTimeoutException generalTimeoutException) {
					log.log(Level.SEVERE, "", generalTimeoutException);
					nodeTransactionPushManager.addBadNode(node);
				} catch (Exception exception) {
					log.log(Level.SEVERE, "", exception);
				}
			}
		}
	}

	public void saveTransactionRecords(String shardid, List<TransactionRecord> transactionRecords) throws Exception {
		if (StringUtils.isBlank(shardid)) {
			throw new Exception("shardid blank");
		}
		if (transactionRecords == null)
			return;
		SearchIndexShard searchIndexShard = searchIndexShardManager.getSearchIndexShard(shardid);
		if (searchIndexShard == null) {
			throw new Exception("unknown shardid: " + shardid);
		}
		searchIndexShard.saveTransactionRecords(transactionRecords, true);
	}

	public List<TransactionRecord> getTransactionRecords(String shardid, TransactionID startid, CLong length) throws Exception {
		if (StringUtils.isBlank(shardid)) {
			throw new Exception("shardid blank");
		}
		if (length == null) {
			throw new Exception("length null");
		}
		SearchIndexShard searchIndexShard = searchIndexShardManager.getSearchIndexShard(shardid);
		if (searchIndexShard == null) {
			throw new Exception("unknown shardid: " + shardid);
		}
		return searchIndexShard.getTransactionRecords(startid, length);
	}

	public Map<String, ShardInfo> getShardInfos() throws Exception {
		Map<String, ShardInfo> map = new HashMap<String, ShardInfo>();
		List<SearchIndexShard> searchIndexShards = searchIndexShardManager.getSearchIndexShards();
		for (SearchIndexShard searchIndexShard : searchIndexShards) {
			ShardInfo shardInfo = searchIndexShard.createShardInfo();
			map.put(shardInfo.id, shardInfo);
		}
		return map;
	}

	public OceanXML.ServerInfo getServerInfo() {
		return new OceanXML.ServerInfo(getID(), "ocean", reportableStatus);
	}

	public TermFrequencies getTermFrequencies(Query query, TimeoutState timeoutState) throws Exception {
		Query rewrittenQuery = rewriteQuery(query, timeoutState);
		MultiShardTermFrequency multiShardTermFrequency = new MultiShardTermFrequency(rewrittenQuery, searchIndexShardManager.getSearchIndexShards(),
				termFrequencyThreadPool, timeoutState);
		return multiShardTermFrequency.getTermFrequencies();
	}

	public Query rewriteQuery(Query original, TimeoutState timeoutState) throws Exception {
		MultiShardQueryRewrite multiRewrite = new MultiShardQueryRewrite(original, searchIndexShardManager.getSearchIndexShards(), queryRewriteThreadPool,
				timeoutState);
		return multiRewrite.rewrite();
	}

	public String getID() {
		return thisNodeID;
	}

	static {
		ClusterUtil.loadStatics();
	}

	static {
		Set<String> idSet = new HashSet<String>();
		idSet.add(ClusterConstants.ID_FIELD);
		idSet.add(ClusterConstants.VERSION_FIELD);
		idFieldSelector = new SetBasedFieldSelector(idSet, null);

		headerSet = new HashSet<String>();
		headerSet.add(ClusterConstants.ID_FIELD);
		headerSet.add(ClusterConstants.TRANSACTION_ID_FIELD);
		headerSet.add(ClusterConstants.VERSION_FIELD);
		headerSet.add(ClusterConstants.DATE_MODIFIED_FIELD);
		headerSet.add(ClusterConstants.DATE_CREATED_FIELD);
		headerFieldSelector = new SetBasedFieldSelector(headerSet, null);

		allFieldSelector = new AllFieldSelector();
	}

	public CurrentIndexSchema getCurrentIndexSchema() {
		return new CurrentIndexSchemaHandler();
	}

	public class CurrentIndexSchemaHandler implements CurrentIndexSchema {
		public IndexSchema getIndexSchema(TimeoutState timeoutState) throws Exception {
			return nodeNameClient.getIndexSchema(timeoutState);
		}

		public IndexSchema getIndexSchema() throws Exception {
			return nodeNameClient.getIndexSchema(new TimeoutState(1000 * 60 * 2));
		}
	}
  // TODO: handle corrupt index reporting to name server
	public void reportCorruptIndex(SearchIndexShard searchIndexShard, CorruptIndexException corruptIndexException) {
		/**
		 * try { NodeShard nodeShard = new NodeShard(); nodeShard.nodeID = getID();
		 * nodeShard.shardID = searchIndexShard.getName(); nodeShard.status =
		 * ClusterConstants.CORRUPT_INDEX; sendNameUpdateNoWait(nodeShard); } catch
		 * (Throwable throwable) { log.log(Level.SEVERE, "", throwable); }
		 */
	}

	/**
	 * public class SearchUDPCommunication extends UDPCommunication2 { public
	 * SearchUDPCommunication(String thisID, int udpPort) throws Exception {
	 * init(thisID, udpPort); } public void sendCellTransactionEvent() throws
	 * Exception { UDPMessage message = new UDPMessage(); message.action =
	 * UDPConstants.CELL_TRANSACTION_EVENT_ACTION; Set<HostPort>
	 * destinationHostPorts = new HashSet<HostPort>(cell.nodes.size()); for (Node
	 * node : cell.nodes) { int port = node.getUDPPort();
	 * destinationHostPorts.add(new HostPort(node.url.getHost(), port)); }
	 * send(destinationHostPorts, message); } public void handleMessage(UDPMessage
	 * message) throws Exception { if
	 * (StringUtils.equals(UDPConstants.NAMES_VOTING_ACTION, message.action)) {
	 * nameClient.resetMaster(); } else if
	 * (StringUtils.equals(UDPConstants.CELL_TRANSACTION_EVENT_ACTION,
	 * message.action)) { replicationClient.notifyNewTransactions(); } } }
	 */

	public String getCellIDNoWait() throws Exception {
		return nodeNameClient.getCellIDNoWait();
	}

	public String getCellID() throws InterruptedException {
		return nodeNameClient.getCellIDWait();
	}

	public boolean isMaster(TimeoutState timeoutState) throws Exception {
		return nodeNameClient.isMaster(timeoutState);
	}

	public boolean isSlave(TimeoutState timeoutState) throws Exception {
		return nodeNameClient.isSlave(timeoutState);
	}

	public static SearchService getSearchService() throws Exception {
		SearchService searchService = (SearchService) ClusterRegistry.get("searchService");
		if (searchService == null) {
			searchService = new SearchService();
			searchService.start();
		}
		return searchService;
	}

	private void errorIfNotUpdateable(TimeoutState timeoutState) throws Exception {
		if (nodeNameClient.isSlave(timeoutState)) {
			throw new ClusterException("not master. not updateable");
		}
	}

	public String loadNodeID() throws Exception {
		return ClusterUtil.loadNodeID(searchConfig.nodeIDFile);
	}

	public SearchService() throws Exception {
		searchConfig = SearchConfig.loadSearchConfig();
	}

	public SearchIndexShard getSearchIndexShard(String shardID) {
		return searchIndexShardManager.getSearchIndexShard(shardID);
	}

	public SearchIndexShardManager getSearchIndexShardManager() {
		return searchIndexShardManager;
	}

	public ThreadPoolExecutor getSearchThreadPool() {
		return searchThreadPool;
	}

	public IndexSchema getIndexSchema() throws Exception {
		return nodeNameClient.getIndexSchema(TimeoutState.FOREVER());
	}

	public SearchResponse search(SearchRequest searchRequest) throws Exception {
		IndexSchema indexSchema = getIndexSchema();
		TimeoutState timeoutState = null;
		if (searchRequest.timeout != null) {
			timeoutState = new TimeoutState(searchRequest.timeout.value);
		} else {
			timeoutState = new TimeoutState();
		}
		String cellID = getCellID(timeoutState);
		// fieldsToLoad.removeAll(defaultDocumentFieldCaches);
		CFieldSelector fieldSelector = null;
		if (searchRequest.fields != null)
			fieldSelector = new CFieldSelector(searchRequest.fields);
		if (searchRequest.lookupIDVersions != null) {
			SearchResponse searchResponse = new SearchResponse();
			searchResponse.scoreDocuments = new ArrayList<CScoreDocument>(searchRequest.lookupIDVersions.size());
			for (IDVersion idVersion : searchRequest.lookupIDVersions) {
				SearchIndexShard searchIndexShard = searchIndexShardManager.getSearchIndexShard(idVersion.shardID);
				Document document = searchIndexShard.getDocument(idVersion, fieldSelector, defaultDocumentFieldCaches);
				searchResponse.scoreDocuments.add(new CScoreDocument(null, document));
			}
			return searchResponse;
		}
		String queryString = searchRequest.queryString;
		Query query = OceanQueryParser.parse(queryString, indexSchema);
		query = QueryUtils.makeQueryable(query);
		Query rewrittenQuery = rewriteQuery(query, timeoutState);
		if (searchRequest.sortFields == null) {
			searchRequest.sortFields = ClusterUtil.toSortFields(searchRequest.sortString);
		}
		List<Query> filterQueries = null;
		if (searchRequest.filterQueryStrings != null) {
			filterQueries = new ArrayList<Query>(searchRequest.filterQueryStrings.size());
			for (String filterQueryString : searchRequest.filterQueryStrings) {
				filterQueries.add(OceanQueryParser.parse(filterQueryString, indexSchema));
			}
		}
		if (searchRequest.distance != null) {
			if (filterQueries == null) {
				filterQueries = new ArrayList<Query>(2);
			}
			ClusterUtil.createDistanceBoxQueries(searchRequest.distance, filterQueries, indexSchema);
		}
		if (searchRequest.termFrequencies == null) {
			searchRequest.termFrequencies = getTermFrequencies(rewrittenQuery, timeoutState);
		}
		// rewrite query
		List<SearchIndexShard> searchIndexShards = searchIndexShardManager.getSearchIndexShards();
		if (searchRequest.rows == null) {
			searchRequest.rows = new CLong(10);
		}
		if (searchRequest.start == null) {
			searchRequest.start = new CLong(0);
		}
		FacetResponse facetResponse = null;
		if (searchRequest.facetRequest != null) {
			facetResponse = new FacetResponse();
			for (FacetRequest.QueryFacet requestFacet : searchRequest.facetRequest.queryFacets) {
				FacetResponse.QueryFacet responseFacet = new FacetResponse.QueryFacet();
				responseFacet.count = new AtomicLong(0);
				responseFacet.id = requestFacet.id;
				responseFacet.queryString = requestFacet.queryString;
				responseFacet.query = OceanQueryParser.parse(requestFacet.queryString, indexSchema);
				facetResponse.queryFacets.add(responseFacet);
			}
			for (FacetRequest.FieldFacet requestFacet : searchRequest.facetRequest.fieldFacets) {
				FacetResponse.FieldFacet responseFacet = new FacetResponse.FieldFacet();
				responseFacet.field = requestFacet.field;
				facetResponse.fieldFacets.add(responseFacet);
			}
		}
		MultiShardSearch multiShardSearch = new MultiShardSearch(rewrittenQuery, searchRequest.start.value, searchRequest.rows.value, filterQueries, facetResponse,
				searchIndexShards, indexSchema.getSimilarity(), timeoutState, searchRequest.termFrequencies, searchThreadPool, indexSchema);
		long startTime = System.currentTimeMillis();
		SearchResponse searchResponse = new SearchResponse();
		searchResponse.start = searchRequest.start;
		searchResponse.rows = searchRequest.rows;
		searchResponse.url = searchRequest.url;
		searchResponse.facetResponse = facetResponse;
		SearchResult searchResult = null;
		Set<String> documentFieldCaches = null;
		if (searchRequest.sortFields == null) {
			searchResult = multiShardSearch.search(searchRequest.distance);
			documentFieldCaches = new HashSet<String>(defaultDocumentFieldCaches.size());
		} else {
			searchResult = multiShardSearch.search(searchRequest.sortFields, searchRequest.distance);
			documentFieldCaches = new HashSet<String>(defaultDocumentFieldCaches.size() + searchRequest.sortFields.size());
			for (CSortField sortField : searchRequest.sortFields) {
				documentFieldCaches.add(sortField.field);
			}
		}
		// TODO: include query in search response
		// documentFieldCaches.addAll(defaultDocumentFieldCaches);
		List<CScoreDocument> scoreDocuments = new ArrayList<CScoreDocument>(searchResult.docs.size());
		for (CScoreDoc scoreDoc : searchResult.docs) {
			Document document = scoreDoc.indexReader.document(scoreDoc.doc, fieldSelector);
			ClusterUtil.setDistance(scoreDoc.distance, document, indexSchema);
			scoreDocuments.add(new CScoreDocument(scoreDoc, document));
		}
		List<OceanDocument> oceanDocuments = ClusterUtil.toOceanDocuments(scoreDocuments, indexSchema);
		searchResponse.numFound = searchResult.numFound;
		//searchResponse.scoreDocuments = scoreDocuments;
		searchResponse.oceanDocuments = oceanDocuments;
		searchResponse.queryString = searchRequest.queryString;
		searchResponse.highlightResponse = doHighlights(searchRequest.highlightRequest, oceanDocuments, rewrittenQuery, indexSchema);
		if (searchResponse != null) {
			long endTime = System.currentTimeMillis();
			searchResponse.queryDuration = new CLong(endTime - startTime);
			searchResponse.cellID = cellID;
			searchResponse.nodeID = getID();
		}
		return searchResponse;
	}

	private HighlightResponse doHighlights(HighlightRequest highlightRequest, List<OceanDocument> oceanDocuments, Query query, IndexSchema indexSchema) throws Exception {
		if (highlightRequest == null)
			return null;
		Highlightings highlightings = new Highlightings(highlightRequest, indexSchema);
		return highlightings.doHighlighting(oceanDocuments, query);
	}

	public UpdateResponse deleteDocumentMaster(IDVersion idVersion, TimeoutState timeoutState) throws Exception {
		idVersion.illegalArgs();
		if (StringUtils.isBlank(idVersion.shardID)) {
			throw new IllegalArgumentException("shardID blank");
		}
		TransactionID cellTransactionID = new TransactionID(cellTransactionIDSequence.next());
		SearchIndexShard searchIndexShard = searchIndexShardManager.getSearchIndexShard(idVersion.shardID);
		return searchIndexShard.deleteDocumentMaster(cellTransactionID, idVersion, timeoutState);
	}

	public void saveTransactionRecord(TransactionRecord transactionRecord) throws Exception {
		if (transactionRecord == null) {
			throw new IllegalArgumentException("transactionRecord null");
		}
		if (transactionRecord.shardID == null) {
			throw new IllegalArgumentException("transactionRecord.shardID null");
		}
		SearchIndexShard searchIndexShard = searchIndexShardManager.getSearchIndexShard(transactionRecord.shardID);
		if (searchIndexShard == null) {
			throw new Exception("unknown shard id: " + transactionRecord.shardID);
		}
		searchIndexShard.saveTransactionRecord(transactionRecord, true);
	}

	public void addDocumentsBatchMaster(String batchID, List<Document> documents, TimeoutState timeoutState) throws Exception {
		SearchIndexShard searchIndexShard = searchIndexShardManager.getNextWriteableIndexShard();
		// return searchIndexShard.addDocumentsBatchMaster(batchID, documents,
		// indexSchema.getAnalyzer(), timeoutState);
		// TODO: finish searchIndexShard.addDocumentsBatchMaster
	}

	public UpdateResponse addDocumentMaster(Document document, TimeoutState timeoutState) throws Exception {
		TransactionID cellTransactionID = new TransactionID(cellTransactionIDSequence.next());
		SearchIndexShard searchIndexShard = searchIndexShardManager.getNextWriteableIndexShard();
		IndexSchema indexSchema = getIndexSchema();
		return searchIndexShard.addDocumentMaster(cellTransactionID, document, indexSchema.getAnalyzer(), timeoutState);
	}

	public String getCellID(TimeoutState timeoutState) throws Exception {
		return nodeNameClient.cellIDRef.get(timeoutState);
	}

	/**
	 * public List<TransactionRecord> getTransactionRecords(List<TransactionRecordRequest>
	 * transactionRecordRequests, CLong rows) throws Exception { List<TransactionRecord>
	 * transactionRecords = new ArrayList<TransactionRecord>(10); for
	 * (TransactionRecordRequest transactionRecordRequest :
	 * transactionRecordRequests) { SearchIndexShard searchIndexShard =
	 * getSearchIndexShardManager().getSearchIndexShard(transactionRecordRequest.shardID);
	 * if (searchIndexShard == null) { } List<TransactionRecord>
	 * shardTransactionRecords =
	 * searchIndexShard.getTransactionRecords(transactionRecordRequest.transactionID,
	 * rows); transactionRecords.addAll(shardTransactionRecords); } return
	 * transactionRecords; }
	 * 
	 * public List<TransactionRecordRequest> getTransactionRecordRequests()
	 * throws Exception { List<SearchIndexShard> searchIndexShards =
	 * getSearchIndexShardManager().getSearchIndexShards(); List<TransactionRecordRequest>
	 * transactionRecordRequests = new ArrayList<TransactionRecordRequest>(searchIndexShards.size());
	 * for (SearchIndexShard searchIndexShard : searchIndexShards) { TransactionID
	 * highestTransactionID = searchIndexShard.getHighestTransactionIDDatabase();
	 * highestTransactionID.increment(); TransactionRecordRequest
	 * transactionRecordRequest = new TransactionRecordRequest();
	 * transactionRecordRequest.shardID = searchIndexShard.getName();
	 * transactionRecordRequest.transactionID = highestTransactionID;
	 * transactionRecordRequests.add(transactionRecordRequest); } return
	 * transactionRecordRequests; }
	 */
	public UpdateResponse updateDocumentMaster(Document document, TimeoutState timeoutState) throws Exception {
		String shardID = ClusterUtil.getShardID(document);
		if (StringUtils.isBlank(shardID)) {
			throw new IllegalArgumentException("shardID blank");
		}
		TransactionID cellTransactionID = new TransactionID(cellTransactionIDSequence.next());
		IndexSchema indexSchema = getIndexSchema();
		SearchIndexShard searchIndexShard = searchIndexShardManager.getSearchIndexShard(shardID);
		return searchIndexShard.updateDocumentMaster(cellTransactionID, document, indexSchema.getAnalyzer(), timeoutState);
	}

	public class AddRAMIndexTask implements Runnable {
		public void run() {
			try {
				List<SearchIndexShard> searchIndexShards = searchIndexShardManager.getSearchIndexShards();
				for (SearchIndexShard searchIndexShard : searchIndexShards) {
					searchIndexShard.reload(new CLong(100));
				}
			} catch (Throwable throwable) {
				log.log(Level.SEVERE, "", throwable);
			}
		}
	}

	public String getServiceStatus() {
		return serviceStatus;
	}

	public class StartThread implements Runnable {
		public StartThread() {
			new Thread(this).start();
		}

		public void run() {
			try {
				serviceStatus = Service.STARTING;
				startInternal();
				serviceStatus = Service.STARTED;
			} catch (Throwable throwable) {
				log.log(Level.SEVERE, "", throwable);
			}
		}
	}

	public void start() throws Exception {
		new StartThread();
	}

	private void startInternal() throws Exception {
		thisNodeID = loadNodeID();
		this.reportableStatus = ClusterConstants.NOTHING;
		cellTransactionIDSequence = new CSequence("celltransactionid", searchConfig.indicesDirectory);
		try {
			FileUtils.cleanDirectory(searchConfig.tempDirectory);
		} catch (Throwable throwable) {
			log.log(Level.SEVERE, "", throwable);
		}
		searchThreadPool = new ThreadPoolExecutor(2, (int) searchConfig.searchThreadPoolSizeMax.value, 5 * 60 * 1000, TimeUnit.MILLISECONDS,
				new LinkedBlockingQueue());
		updateThreadPool = new ThreadPoolExecutor(1, (int) searchConfig.updateThreadPoolSizeMax.value, 5 * 60 * 1000, TimeUnit.MILLISECONDS,
				new LinkedBlockingQueue());
		queryRewriteThreadPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(10);
		termFrequencyThreadPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(10);
		timer = Executors.newScheduledThreadPool(1);

		String thisHost = searchConfig.url.getHost();
		Set<URL> nameURLs = ClusterUtil.loadURLs(searchConfig.nameURLsFile);
		log.info("node id: " + getID());
		log.info("nameURLs: " + nameURLs);
		nodeNameClient = new NodeNameClient(getID(), nameURLs, new NodeEventHandler());
		log.log(Level.INFO, "waiting for IndexSchema");
		IndexSchema indexSchema = nodeNameClient.getIndexSchema(TimeoutState.FOREVER());
		log.log(Level.INFO, "obtained IndexSchema");

		searchIndexShardManager = new SearchIndexShardManager(this);

		nodeNameClient.startNodeNameClient();

		//getShardInfosTimer = new RestartableTimer(new GetShardInfosTask(), 1000 * 5, 1000 * 60);
		replicationClient = new ReplicationClient(new ReplicationHandler());
		checkMasterSlaveStatus(true);
		timer.scheduleWithFixedDelay(new AddRAMIndexTask(), ADD_RAM_INDEX_DELAY, ADD_RAM_INDEX_DELAY, TimeUnit.MILLISECONDS);
		log.log(Level.INFO, "Ocean Search Service started...");
	}
  // TODO: master changes reportable status to nothing while other slave is getting started
	// TODO: reported status since date
	/**
	 * Returns immediately if the lock is currently held
	 */
	public void checkMasterSlaveStatus(boolean waitForMaster) throws Exception {
		if (!checkMasterSlaveStatusLock.tryLock(25, TimeUnit.MILLISECONDS)) {
			return;
		}
		try {
			Node masterNode = null;
			if (waitForMaster) masterNode = nodeNameClient.masterNodeRef.get();
			else masterNode = nodeNameClient.masterNodeRef.getNoWait();
			if (masterNode == null && !StringUtils.equals(reportableStatus, ClusterConstants.MASTER))  {
				reportableStatus = ClusterConstants.NOTHING;
				return;
			}
			//log.info("masterNode: " + masterNode);
			if (StringUtils.equals(getID(), masterNode.id)) {
				if (reportableStatus == null || !StringUtils.equals(ClusterConstants.MASTER, reportableStatus)) {
					log.info("i'm master");
					searchIndexShardManager.createMinimumNumberOfShards();
					replicationClient.stopReplicating();
					resetSequences();
					searchIndexShardManager.resetSequences();
					reportableStatus = ClusterConstants.MASTER;
				}
			} else {
				if (reportableStatus == null 
						|| !StringUtils.equals(ClusterConstants.SLAVE, reportableStatus)) {
					log.info("i'm slave");
					//getShardInfosTimer.runNow();
					replicationClient.startReplicating(masterNode.url);
					reportableStatus = ClusterConstants.SLAVE;
				}
			}
		} finally {
			checkMasterSlaveStatusLock.unlock();
		}
	}

	public void triggerEvent(String name) {
		try {
			if (StringUtils.equals("newshard", name)) {
				//getShardInfosTimer.runNow();
				replicationClient.notifyNewTransactions();
			} else if (StringUtils.equals("newtransactions", name)) {
				replicationClient.notifyNewTransactions();
			}
		} catch (Throwable throwable) {
			log.log(Level.SEVERE, "", throwable);
		}
	}

	// TODO: if transaction is pushed or read and the shard is unknown call this
	// to make sure it's created
	// TODO: put this in the replication client
	/**
	public class GetShardInfosTask implements RestartableTimer.TimeoutRunnable {
		public void run(TimeoutState timeoutState) {
			try {
				Node masterNode = nodeNameClient.masterNodeRef.getNoWait();
				if (masterNode != null && !StringUtils.equals(getID(), masterNode.id)) {
					HttpTimeout httpTimeout = new HttpTimeout(1000 * 3, 1000 * 3, timeoutState);
					Map<String, ShardInfo> shardInfos = NodeClient.getShardInfos(masterNode.url, httpTimeout);
					searchIndexShardManager.createShards(shardInfos);
				}
			} catch (Throwable throwable) {
				log.log(Level.SEVERE, "", throwable);
			}
		}
	}
  **/
	public class ReplicationHandler implements ReplicationClient.ReplicationInterface {
		public void ensureShardsExist(Map<String, ShardInfo> shardInfos) throws Exception {
			searchIndexShardManager.createShards(shardInfos);
		}
		
		public TransactionID getHighestTransactionIDDatabase(String shardID) {
			try {
				SearchIndexShard searchIndexShard = searchIndexShardManager.getSearchIndexShard(shardID);
				if (searchIndexShard == null)
					return null;
				return searchIndexShard.getHighestTransactionIDDatabase();
			} catch (Throwable throwable) {
				log.log(Level.SEVERE, "", throwable);
				return null;
			}
		}

		public void transactions(String shardID, List<TransactionRecord> transactionRecords) {
			SearchIndexShard searchIndexShard = searchIndexShardManager.getSearchIndexShard(shardID);
			if (searchIndexShard == null) {
				log.log(Level.SEVERE, "unknown shardID: " + shardID);
			} else {
				try {
					searchIndexShard.saveTransactionRecords(transactionRecords, true);
				} catch (Throwable throwable) {
					log.log(Level.SEVERE, "", throwable);
				}
			}
		}
	}

	public String getReportableStatus() {
		return reportableStatus;
	}

	public Node createReportableNode() {
		Node node = new Node();
		node.id = getID();
		node.reportedStatus = getReportableStatus();
		if (nodeNameClient.myChosenMaster != null) {
			node.reportedMasterID = nodeNameClient.myChosenMaster.id;
		}
		node.reportedCellID = nodeNameClient.cellIDRef.getNoWait();
		try {
			node.highestCellTransactionIDDatabase = getHighestCellTransactionIDDatabase();
		} catch (Throwable throwable) {
			log.log(Level.SEVERE, "", throwable);
		}
		node.url = searchConfig.url;
		return node;
	}

	public class NodeEventHandler extends NodeNameClient.NodeEventListener {
		public Node getReportableNode() {
			return createReportableNode();
		}

		public void masterNodeChanged() {
			try {
				Node masterNode = nodeNameClient.masterNodeRef.getNoWait();
				//log.log(Level.INFO, "master changed event node: " + masterNode);
				//if (masterNode != null) {
				if (!SearchService.this.checkMasterSlaveStatusLock.isLocked()) {
			    checkMasterSlaveStatus(false);
				}
				//} else {
				//	log.info("no new master");
				//}
			} catch (Throwable throwable) {
				log.log(Level.SEVERE, "", throwable);
			}
		}
	}

	public void stop() throws Exception {
		// changeStatus(ClusterConstants.NOT_AVAILABLE);
		log.info("SearchService.stop");
		serviceStatus = Service.STOPPING;
		try {
			if (replicationClient != null) {
				replicationClient.shutdown();
			}
		} catch (Throwable ex) {
			log.log(Level.SEVERE, "", ex);
		}
		try {
			if (nodeNameClient != null) {
				nodeNameClient.shutdown();
			}
		} catch (Throwable ex) {
			log.log(Level.SEVERE, "", ex);
		}
		try {
			if (timer != null) {
				timer.shutdown();
			}
		} catch (Throwable ex) {
			log.log(Level.SEVERE, "", ex);
		}
		try {
			if (searchThreadPool != null) {
				searchThreadPool.shutdown();
			}
		} catch (Throwable ex) {
			log.log(Level.SEVERE, "", ex);
		}
		try {
			if (updateThreadPool != null) {
				updateThreadPool.shutdown();
			}
		} catch (Throwable ex) {
			log.log(Level.SEVERE, "", ex);
		}
		try {
			if (queryRewriteThreadPool != null) {
				queryRewriteThreadPool.shutdown();
			}
		} catch (Throwable ex) {
			log.log(Level.SEVERE, "", ex);
		}
		try {
			if (termFrequencyThreadPool != null) {
				termFrequencyThreadPool.shutdown();
			}
		} catch (Throwable ex) {
			log.log(Level.SEVERE, "", ex);
		}
		try {
			if (searchIndexShardManager != null) {
				searchIndexShardManager.close();
			}
		} catch (Throwable ex) {
			log.log(Level.SEVERE, "", ex);
		}
		serviceStatus = Service.STOPPED;
	}

	public class ExitThread extends Thread {
		public ExitThread() {
			this.setDaemon(true);
			start();
		}

		public void run() {
			try {
				InputStreamReader reader = new InputStreamReader(System.in);
				while (true) {
					char ch = (char) reader.read();
					if (ch == 'q') {
						SearchService.this.stop();
						System.exit(0);
						return;
					}
				}
			} catch (Throwable throwable) {
				log.log(Level.SEVERE, "", throwable);
			}
		}
	}
}
